use std::net::{SocketAddr, TcpListener};
use std::time::Duration;

use flate2::write::GzEncoder;
use flate2::Compression;
use pdata::internal::data::protogen::collector::trace::v1::ExportTraceServiceRequest;
use pdata::internal::data::protogen::trace::v1::{ResourceSpans, ScopeSpans, Span};
use prost::Message;
use reqwest::Client;
use rust_collector::receiver::otlp_http_receiver::OtlpHttpReceiver;
use tokio::time::sleep;

fn pick_free_port() -> u16 {
    let l = TcpListener::bind("127.0.0.1:0").expect("bind temp");
    let port = l.local_addr().unwrap().port();
    drop(l);
    port
}

fn minimal_trace_request() -> ExportTraceServiceRequest {
    let span = Span {
        trace_id: vec![0; 16],
        span_id: vec![0; 8],
        trace_state: String::new(),
        parent_span_id: Vec::new(),
        flags: 0,
        name: "test_span".to_string(),
        kind: 1,
        start_time_unix_nano: 1,
        end_time_unix_nano: 2,
        attributes: Vec::new(),
        dropped_attributes_count: 0,
        events: Vec::new(),
        dropped_events_count: 0,
        links: Vec::new(),
        dropped_links_count: 0,
        status: None,
    };

    let scope_spans = ScopeSpans { scope: None, spans: vec![span], schema_url: String::new() };
    let resource_spans = ResourceSpans { resource: None, scope_spans: vec![scope_spans], schema_url: String::new() };
    ExportTraceServiceRequest { resource_spans: vec![resource_spans] }
}

async fn start_http_receiver(addr: SocketAddr) {
    let recv = OtlpHttpReceiver::new(None, None, None);
    tokio::spawn(async move {
        let _ = recv.run(addr).await;
    });
    // allow server to come up
    sleep(Duration::from_millis(120)).await;
}

#[tokio::test]
async fn post_traces_ok() {
    let port = pick_free_port();
    let addr: SocketAddr = format!("127.0.0.1:{}", port).parse().unwrap();
    start_http_receiver(addr).await;

    let client = Client::new();
    let url = format!("http://{}/v1/traces", addr);

    let req = minimal_trace_request();
    let mut body = Vec::new();
    req.encode(&mut body).unwrap();

    let resp = client
        .post(&url)
        .header(reqwest::header::CONTENT_TYPE, "application/x-protobuf")
        .header(reqwest::header::ACCEPT, "application/x-protobuf")
        .body(body)
        .send()
        .await
        .unwrap();
    assert!(resp.status().is_success());
}

#[tokio::test]
async fn post_traces_gzip_ok() {
    let port = pick_free_port();
    let addr: SocketAddr = format!("127.0.0.1:{}", port).parse().unwrap();
    start_http_receiver(addr).await;

    let client = Client::new();
    let url = format!("http://{}/v1/traces", addr);

    let req = minimal_trace_request();
    let mut plain = Vec::new();
    req.encode(&mut plain).unwrap();
    let mut enc = GzEncoder::new(Vec::new(), Compression::default());
    std::io::Write::write_all(&mut enc, &plain).unwrap();
    let gz = enc.finish().unwrap();

    let resp = client
        .post(&url)
        .header(reqwest::header::CONTENT_TYPE, "application/x-protobuf")
        .header(reqwest::header::ACCEPT, "application/x-protobuf")
        .header(reqwest::header::CONTENT_ENCODING, "gzip")
        .body(gz)
        .send()
        .await
        .unwrap();
    assert!(resp.status().is_success());
}

#[tokio::test]
async fn invalid_protobuf_returns_400() {
    let port = pick_free_port();
    let addr: SocketAddr = format!("127.0.0.1:{}", port).parse().unwrap();
    start_http_receiver(addr).await;

    let client = Client::new();
    let url = format!("http://{}/v1/traces", addr);

    let resp = client
        .post(&url)
        .header(reqwest::header::CONTENT_TYPE, "application/x-protobuf")
        .header(reqwest::header::ACCEPT, "application/x-protobuf")
        .body(vec![1, 2, 3, 4, 5])
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), reqwest::StatusCode::BAD_REQUEST);
}

#[tokio::test]
async fn not_found_returns_404() {
    let port = pick_free_port();
    let addr: SocketAddr = format!("127.0.0.1:{}", port).parse().unwrap();
    start_http_receiver(addr).await;

    let client = Client::new();
    let url = format!("http://{}/v1/unknown", addr);
    let resp = client
        .post(&url)
        .body(Vec::<u8>::new())
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), reqwest::StatusCode::NOT_FOUND);
}

#[tokio::test]
async fn method_not_allowed_returns_405() {
    let port = pick_free_port();
    let addr: SocketAddr = format!("127.0.0.1:{}", port).parse().unwrap();
    start_http_receiver(addr).await;

    let client = Client::new();
    let url = format!("http://{}/v1/traces", addr);
    let resp = client.get(&url).send().await.unwrap();
    assert_eq!(resp.status(), reqwest::StatusCode::METHOD_NOT_ALLOWED);
}


